package cn.enjoy.kafka.controller.boot;

import cn.enjoy.kafka.constant.KafkaConstant;
import cn.enjoy.kafka.entity.City;
import cn.enjoy.tool.api.R;
import com.alibaba.fastjson2.JSONObject;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;

/**
 * @author K
 * @date 2024-06-30 15:01
 */
@Slf4j
@RestController
@RequestMapping("producer")
public class ProducerController {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;


    @ApiOperation("异步发送")
    @PostMapping("/asyncSend")
    public R asyncSend() {
        try {
            City city = new City();
            city.setCityId(2023064321L);
            city.setCityName("成都");
            String topicName = KafkaConstant.TOPIC_BOOT_TEST;
            String msgKey = System.currentTimeMillis() + "";
            String msgData = JSONObject.toJSONString(city);
            log.info("向Kafka集群发送消息，Topic:{} Key:{} Data:{}", topicName, msgKey, msgData);
            kafkaTemplate.send(topicName, msgKey, msgData);
        } catch (Exception e) {
            log.info("向Kafka集群发送消息异常", e);
        }
        return R.success("成功");
    }

    @ApiOperation("异步发送-带回调")
    @PostMapping("/asyncSendCallback")
    public R asyncSendCallback() {
        try {
            City city = new City();
            city.setCityId(2023064321L);
            city.setCityName("成都");
            String topicName = KafkaConstant.TOPIC_BOOT_TEST;
            String msgKey = System.currentTimeMillis() + "";
            String msgData = JSONObject.toJSONString(city);
            log.info("向Kafka集群发送消息，Topic:{} Key:{} Data:{}", topicName, msgKey, msgData);
            kafkaTemplate.send(topicName, msgKey, msgData).addCallback(success -> {
                // 消息发送到的topic
                String topic = success.getRecordMetadata().topic();
                // 消息发送到的分区
                int partition = success.getRecordMetadata().partition();
                // 消息在分区内的offset
                long offset = success.getRecordMetadata().offset();
                log.info("发送消息成功:" + topic + "-" + partition + "-" + offset + "-" + msgData);
            }, failure -> {
                log.info("发送消息失败:" + failure.getMessage());
            });

        } catch (Exception e) {
            log.info("向Kafka集群发送消息异常", e);
        }
        return R.success("成功");
    }

    @ApiOperation("异步发送-带回调")
    @PostMapping("/asyncSendCallback2")
    public R asyncSendCallback2() {
        try {
            City city = new City();
            city.setCityId(2023064321L);
            city.setCityName("成都");
            String topicName = KafkaConstant.TOPIC_BOOT_TEST;
            String msgKey = System.currentTimeMillis() + "";
            String msgData = JSONObject.toJSONString(city);
            log.info("向Kafka集群发送消息，Topic:{} Key:{} Data:{}", topicName, msgKey, msgData);
            kafkaTemplate.send(topicName, msgKey, msgData).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

                @Override
                public void onSuccess(SendResult<String, String> result) {
                    RecordMetadata rm = result.getRecordMetadata();
                    log.info("发送消息成功：" + rm.topic() + "-" + rm.partition() + "-" + rm.offset() + "-" + msgData);
                }

                @Override
                public void onFailure(Throwable ex) {
                    log.info("发送消息失败：" + ex.getMessage());
                }
            });

        } catch (Exception e) {
            log.info("向Kafka集群发送消息异常", e);
        }
        return R.success("成功");
    }

    @ApiOperation("同步发送")
    @PostMapping("/syncSend")
    public R syncSend() {
        try {
            City city = new City();
            city.setCityId(2023064321L);
            city.setCityName("成都");
            String topicName = KafkaConstant.TOPIC_BOOT_TEST;
            String msgKey = System.currentTimeMillis() + "";
            String msgData = JSONObject.toJSONString(city);
            log.info("向Kafka集群发送消息，Topic:{} Key:{} Data:{}", topicName, msgKey, msgData);
            kafkaTemplate.send(topicName, msgKey, msgData).get();
        } catch (Exception e) {
            log.info("向Kafka集群发送消息异常", e);
        }
        return R.success("成功");
    }

    @PostMapping("/send")
    public void sendMessage(@RequestBody String message) {
        kafkaTemplate.send(KafkaConstant.TOPIC_BOOT_MY, message);
    }

}
